package com.tang.sink.custom;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

/**
 * 自定义Sink
 *
 * @author tang
 * @since 2023/6/9 18:01
 */
public class MySink<In> extends RichSinkFunction<In> {

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        // 在这里创建连接
        // conn = new xxxx
    }

    @Override
    public void close() throws Exception {
        super.close();
        // 做一些清理、销毁连接
    }

    /**
     * Writes the given value to the sink. This function is called for every record.
     *
     * <p>You have to override this method when implementing a {@code SinkFunction}, this is a
     * {@code default} method for backward compatibility with the old-style method only.
     *
     * @param value   The input record.
     * @param context Additional context about the input record.
     * @throws Exception This method may throw exceptions. Throwing an exception will cause the
     *                   operation to fail and may trigger recovery.
     */
    @Override
    public void invoke(In value, Context context) throws Exception {
        // 写出逻辑实现，这个方法是来一条数据就会调用一次所以不要在这里连接对象。
        System.out.println(value);
    }
    
}
